package org.budo.ehcache.replicator.redis;

import org.budo.ehcache.replicator.EhcacheReplicator;
import org.budo.ehcache.replicator.EhcacheReplicatorFactory;
import org.budo.ehcache.replicator.ReplicatorMessage;
import org.budo.support.assertion.Assert;
import org.budo.support.lang.util.StringUtil;
import org.springframework.beans.factory.InitializingBean;

import com.alibaba.fastjson.JSON;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPubSub;

/**
 * EhcacheRedisReplicator 利用Redis时间通知来同步分布式缓存的监听器
 * 
 * @author lmw
 * @see org.budo.ehcache.replicator.EhcacheReplicatorFactory
 */
@Slf4j
@Getter
@Setter
public class EhcacheRedisReplicator extends JedisPubSub implements EhcacheReplicator, InitializingBean {
    private static final ThreadLocal<String> _LOCAL_MESSAGE = new ThreadLocal<String>();

    private JedisPool jedisPool;

    /**
     * 分布式同步事件通知频道
     */
    private String redisChannel = "EhcacheRedisReplicator"; // 通知频道默认值

    /**
     * 兼容老版本，已经未适用
     * 
     * @deprecated
     */
    @Deprecated
    public void setCacheManager(Object cacheManager) {
    }

    /**
     * 启动监听线程
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        Thread checkerThread = new Thread() {
            public void run() {
                while (true) {
                    Jedis jedis = null;
                    try {
                        jedis = EhcacheRedisReplicator.this.getJedis();
                        jedis.subscribe(EhcacheRedisReplicator.this, EhcacheRedisReplicator.this.getRedisChannel());
                    } catch (Throwable e) {
                        log.error("#53 subscribeRedis error, e=" + e, e);
                    } finally {
                        EhcacheRedisReplicator.this.returnJedis(jedis);
                    }
                }
            }
        };

        checkerThread.setName("EhcacheRedisReplicator-RedisSubscribeThread");

        checkerThread.start();
    }

    /**
     * JedisPubSub#onMessage
     */
    @Override
    public void onMessage(String channel, String message) {
        log.debug("#69 onMessage, channel={}, message={}, jedisPool={}", channel, message, this.getJedisPool());

        ReplicatorMessage eventMessage = JSON.parseObject(message, ReplicatorMessage.class);

        try {
            _LOCAL_MESSAGE.set("this=" + this + ", message=" + message); // 在当前线程上标记，此次是收到同步事件后做的操作

            EhcacheReplicatorFactory.onReplicatorMessage(eventMessage, this);
        } catch (Throwable e) {
            throw e instanceof RuntimeException ? (RuntimeException) e : new RuntimeException(e);
        } finally {
            _LOCAL_MESSAGE.remove(); // 清理线程上的标记
        }
    }

    /**
     * 发送同步信号，如果是收到同步事件后做的操作，则不再次发送同步事件
     */
    @Override
    public void publishReplicatorMessage(ReplicatorMessage message) {
        log.debug("#89 publishReplicatorMessage, message={}, jedisPool={}", message, this.getJedisPool());

        if (!StringUtil.isEmpty(_LOCAL_MESSAGE.get())) {
            return; // 收到同步信号后做出的操作
        }

        Jedis jedis = null;
        try {
            jedis = this.getJedis();

            jedis.publish(this.getRedisChannel(), JSON.toJSONString(message));
        } catch (Throwable e) {
            log.error("#117 publishRedisMessage error, e=" + e, e);
        } finally {
            this.returnJedis(jedis);
        }
    }

    protected Jedis getJedis() {
        JedisPool jedisPool = this.getJedisPool();
        Assert.notNull(jedisPool, "#125 jedisPool is null, this=" + this);

        return jedisPool.getResource();
    }

    protected void returnJedis(Jedis jedis) {
        if (null != jedis) {
            jedis.close();
        }
    }
}